package com.czk.rdd.persist

import org.apache.spark.{SparkConf, SparkContext}

/**
 * chechepoint
 * checkPoint与cache persist的区别：
 * 1）Cache 缓存只是将数据保存起来，不切断血缘依赖。Checkpoint 检查点切断血缘依赖。
 * 2）Cache 缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint 的数据通常存
 * 储在 HDFS 等容错、高可用的文件系统，可靠性高。
 * 3）建议对 checkpoint()的 RDD 使用 Cache 缓存，这样 checkpoint 的 job 只需从 Cache 缓存
 * 中读取数据即可，否则需要再从头计算一次 RDD
 * 4 persist 设计磁盘IO，性能较低，作业执行完毕后，临时文件会消失
 * @Author:ChenZhangKun
 * @Date: 2021/12/13 10:30
 */
object PersisitDemo02 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    // 一般保存路径是在分布式存储系统中
    sc.setCheckpointDir("cp")

    val list = List("hello scala", "hello, spark")

    val rdd = sc.makeRDD(list)

    val flatRDD = rdd.flatMap(_.split(" "))
    // map可以执行持久化操作
    val mapRDD = flatRDD.map(word=>{
      println("xxxxxxxxxxxx")
      // 返回数据
      (word,1)
    })
    // checkPoint 需要落盘，需要指定检查点保存路径
    mapRDD.checkpoint()
    val reduceRDD = mapRDD.reduceByKey(_ + _)
    reduceRDD.collect().foreach(println)
  }
}
